package org.zn.note.java8;

import java.util.Random;
import java.util.concurrent.*;
import java.util.function.Supplier;
import java.util.stream.Stream;

/**
 * 组合式异步编程
 * <p>
 * 并发：【CPU-1 task1 - task2 - task1】【CPU-2 空】
 * 并行：【CPU-1 task1】【CPU-2 task2】
 * <p>
 * Future
 * Java 5 引入的，用于建模一种异步计算，返回一个执行运算结果的引用，使线程不用阻塞可以继续执行。等用到这个结果的时候再判断、使用。
 * 如何使用？
 * {@link ExecutorService#submit(Callable)}，原来是用{@link Runnable}，现在是实现{@link Callable}。会返回{@link Future}。
 * 例子 {@link #testFuture()}
 * <p>
 * CompletableFuture
 * implements Future<T>，用起来更方便
 * 例子 {@link Shop#getPriceAsync}、{@link Shop#getPriceAsync2}  有点像异步编程了
 *
 * 用 {@link Stream#parallel()} 还是 {@link CompletableFuture#supplyAsync(Supplier)} ？
 * 1、计算密集型的，用Stream并行这种
 * 2、IO等待型的，用CompletableFuture这种
 */
public class Ch11_CompletableFuture {

    public static void main(String[] args) {
        testFuture();
    }

    private static void testFuture() {
        // 创建线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        // 提交一个任务，并获取计算结果引用future
        Future<Double> future = executor.submit(new Callable<Double>() {
            public Double call() {
                return doSomeLongComputation();
            }
        });
        executor.shutdown();
        // 非阻塞，可以做一些其他事情
        doSomethingElse();
        // 获取异步计算的结果
        try {
            // 获取不到结果，result = null
            Double result = future.get(1, TimeUnit.SECONDS); // 获取结果，最多等1秒
            System.out.println(result);
        } catch (ExecutionException ee) { // 计算抛出一个异常
        } catch (InterruptedException ie) { // 当前线程在等待过程中被中断
        } catch (TimeoutException te) { // 在Future对象完成之前超过已过期
        }
    }

    private static Double doSomeLongComputation() {
        return null;
    }

    private static Double doSomethingElse() {
        return null;
    }

    // 在线商店：实现异步API
    private static class Shop {

        private static final Random random = new Random();

        // 对外的接口：根据商品名，查询价格
        // 目前是同步的，调用方会被卡主，等待查询结果
        public double getPrice(String product) {
            return calculatePrice(product);
        }

        // 异步的对外查询接口
        // 在另外一个线程中查询，当前线程直接返回结果的引用
        public Future<Double> getPriceAsync(String product) {
            CompletableFuture<Double> futurePrice = new CompletableFuture<>();
            new Thread(() -> {
                double price = calculatePrice(product);
                futurePrice.complete(price);
            }).start();
            return futurePrice;
        }

        public Future<Double> getPriceAsync2(String product) {
            // 会默认使用ForkJoinPool
            // 同时有重载方法，可以支持传入自己的线程池Executor来执行
            return CompletableFuture.supplyAsync(() -> calculatePrice(product));
        }

        // 计算价格
        private double calculatePrice(String product) {
            // 这里可能会查数据库，可能会调用其他厂商的API，总之就是有个耗时的过程，用delay模拟
            delay();
            return random.nextDouble() * product.charAt(0) + product.charAt(1);
        }

        // 延迟1秒
        private static void delay() {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
